/*
 * Copyright (c) 2022. China Mobile (SuZhou) Software Technology Co.,Ltd. All rights reserved.
 * Lakehouse is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */

package com.chinamobile.cmss.lakehouse.engine.spark;

import com.chinamobile.cmss.lakehouse.common.engine.DirectEngine;
import com.chinamobile.cmss.lakehouse.common.engine.EngineLauncher;
import com.chinamobile.cmss.lakehouse.common.utils.EngineUtils;

import java.io.File;
import java.util.ArrayList;
import java.util.List;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class SparkEngine extends EngineLauncher<SparkCommandBuilder> implements DirectEngine {

    public SparkEngine(SparkCommandBuilder builder) {
        super(builder);
    }

    public SparkEngine setSparkHome(String sparkHome) {
        this.builder.childEnv().put("SPARK_HOME", sparkHome);
        return this;
    }

    @Override
    protected String findSubmitShell() throws Exception {
        return String.join(File.separator, new String[] {
            EngineUtils.getEngineHome("SPARK_HOME", this.builder.childEnv()), "bin", "spark-submit"});
    }

    @Override
    public Process submit() {
        return launcher();
    }

    @Override
    public Process submit(File logFile) {
        return launcher(logFile);
    }

    @Override
    public String processLog() {
        return null;
    }

    @Override
    public void kill(String podId) {
        try {
            List<String> cmd = new ArrayList<>();
            cmd.add(findSubmitShell());
            cmd.addAll(getKillArgs(podId));
            ProcessBuilder pb = createBuilder(cmd);
            pb.start();
        } catch (Exception e) {
            log.error("Failed to process kill command!", e);
        }
    }

    private List<String> getKillArgs(String podId) {
        return builder.buildKillArgs(podId);
    }
}
